package com.hzgj.bcl.stats;

import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.collect.Lists;
import com.hzgj.bcl.redis.RedisClient;
import com.hzgj.bcl.soa.monitor.MonitorConfig;
import com.hzgj.bcl.soa.monitor.MonitorService;
import com.hzgj.bcl.soa.registry.ServiceNotExistException;
import com.hzgj.bcl.soa.stats.Stats;
import com.hzgj.bcl.soa.stats.StatsCollectors;
import com.hzgj.bcl.soa.stats.StatsData;
import com.hzgj.bcl.util.IocContext;
import com.hzgj.bcl.util.jmx.annotation.MBean;
import com.hzgj.bcl.util.jmx.annotation.ManagedOperation;
import com.hzgj.bcl.util.lang.ThreadPoolKit;
import lombok.Builder;
import lombok.Setter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;

import java.util.List;
import java.util.Objects;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Created by Administrator on 2016/8/15.
 */
@MBean(description = "RemoteSendStats")
public class RemoteSendStats implements Stats {

    public static final int queueSize = 50000;
    private Logger logger = LoggerFactory.getLogger(getClass());
    LinkedBlockingQueue<StatsData> queue = new LinkedBlockingQueue<>(queueSize);
    ScheduledExecutorService scheduledExecutorService;
    @Setter
    static long intervalSpan = 1;//默认5分钟
    RedisClient redisClient;
    private AtomicBoolean tick_running = new AtomicBoolean(false);

    public Cache<MetricKey, Metric> cache =
            CacheBuilder.newBuilder()
                    .expireAfterAccess(intervalSpan * 2, TimeUnit.MINUTES)
                    .maximumSize(1000L)
                    .build();

    @Override
    public boolean enabled() {
        return MonitorConfig.getInstance().getMonitorService() != null;
    }

    @ManagedOperation
    public long getQueueSize() {
        return queue.size();
    }

    @Override
    public void report(StatsData data) {
        queue.offer(data);
    }

    @Override
    public String getName() {
        return "remote";
    }

    @Override
    public void start() {
        final MonitorService monitorService = MonitorConfig.getInstance().getMonitorService();
        if (monitorService == null) {
            StatsCollectors.getInstance().disable(getName());
            logger.info("disable remoteSendStats");
            return;
        }
        scheduledExecutorService = ThreadPoolKit.getScheduledExecutorService("remote SendStats");
        final AtomicBoolean handle = new AtomicBoolean(false);
        scheduledExecutorService.scheduleWithFixedDelay(() -> {
            if (queue.size() > 0 && !handle.get()) {
                List<StatsData> data = Lists.newArrayList();
                queue.drainTo(data, 1000);
                if (data.size() > 0) {
                    handle.set(true);
                    try {
                        startTick();
                        monitorService.handle(data);
                    } catch (Exception e) {
                        logger.error("{}", e);
                        if (e instanceof ServiceNotExistException) {
                            try {
                                //监控不存在
                                TimeUnit.SECONDS.sleep(5);
                            } catch (InterruptedException e1) {
                            }
                        }
                    }
                    try {
                        handleLocalDataset(data);
//                    outRemainderValue();
                    } catch (Exception e) {
                        logger.error("{}", e);
                    }
                    handle.set(false);
                }
            }
        }, 0, 3, TimeUnit.SECONDS);
        redisClient = IocContext.applicationContext.getBean(RedisClient.class);
        Assert.notNull(redisClient, "can't find a redisClient");
    }

    private void startTick() {
        if (tick_running.get()) {
            return;
        }
        if (tick_running.compareAndSet(false, true)) {
            ScheduledExecutorService tickExecutor = ThreadPoolKit.getScheduledExecutorService("remote send stats tick");
            tickExecutor.scheduleAtFixedRate(() -> {
                StatsData statsData = new StatsData();
                statsData.type = StatsConst.STATS_TICK_TYPE;
                statsData.time = System.currentTimeMillis();
                queue.add(statsData);
            }, intervalSpan, intervalSpan, TimeUnit.MINUTES);
            logger.info("start remote stats");
        }
    }

    private void outRemainderValue() {
        cache.asMap().forEach((key, metric) -> {
            if (metric.hasData()) {
                TimeFlowCounter timeFlowCounter = (TimeFlowCounter) metric.getTimeCounter();
                StatsHandler.handleDataWithRedis(redisClient, getLocalRedisKey(key), metric, timeFlowCounter.getThreshold().get());
                redisClient.sadd(getNodeRedisKey(key), getNode(key));
            }
        });
    }

    private void handleLocalDataset(List<StatsData> statsDatas) {
        statsDatas.forEach(statsData -> handleLocalData(statsData));
    }

    private void handleLocalData(StatsData statsData) {
        try {

            if (StatsConst.STATS_TICK_TYPE.equals(statsData.type)) {
                cache.asMap().keySet().forEach(key -> {
                    Metric metric = cache.getIfPresent(key);
                    if (metric != null) {
                        metric.updateTick(statsData.time);
                    }
                });
            } else {
                MetricKey metricKey = MetricKey.builder().dstHost(statsData.dstHost).dstPort(statsData.dstPort).service(statsData.service).method(statsData.method).build();
                Metric metric = cache.get(metricKey, () -> new Metric(0, new TimeFlowCounter(intervalSpan,
                        (innerMetric, time) -> {
                            StatsHandler.handleDataWithRedis(redisClient, getLocalRedisKey(statsData), innerMetric, time);
                            redisClient.sadd(getNodeRedisKey(statsData), getNode(statsData));
                            redisClient.set(getNode(statsData) + "_" + RedisKey.UPDATE_TIME, System.currentTimeMillis() + "");
                            innerMetric.clean();
                        })));
                metric.update(statsData.span, statsData.time, statsData.success);
            }
        } catch (ExecutionException e) {
            logger.error("{}", e);
        }
    }

    @Override
    public void stop() {
        scheduledExecutorService.shutdown();
    }

    private String getLocalRedisKey(StatsData statsData) {
        return statsData.service + "_" + statsData.method + "_" + statsData.dstHost + "_" + statsData.dstPort;
    }

    private String getNodeRedisKey(StatsData statsData) {
        return statsData.service + "_" + RedisKey.NODE;
    }

    private String getNodeRedisKey(MetricKey metricKey) {
        return metricKey.service + "_" + RedisKey.NODE;
    }

    private String getNode(StatsData statsData) {
        return statsData.dstHost + ":" + statsData.dstPort;
    }

    private String getNode(MetricKey metricKey) {
        return metricKey.dstHost + ":" + metricKey.dstPort;
    }

    private String getLocalRedisKey(MetricKey metricKey) {
        return metricKey.service + "_" + metricKey.method + "_" + metricKey.dstHost + "_" + metricKey.dstPort;
    }

    @Builder
    static class MetricKey {
        public String dstHost;
        public Integer dstPort;
        public String service;
        public String method;

        public String getKey() {
            return service + "_" + method;
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) {
                return true;
            }
            if (o == null || getClass() != o.getClass()) {
                return false;
            }
            MetricKey metricKey = (MetricKey) o;
            return Objects.equals(dstHost, metricKey.dstHost) &&
                    Objects.equals(dstPort, metricKey.dstPort) &&
                    Objects.equals(service, metricKey.service) &&
                    Objects.equals(method, metricKey.method);
        }

        @Override
        public int hashCode() {
            return Objects.hash(dstHost, dstPort, service, method);
        }
    }
}
